package com.rk.topN;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class CombineDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        String countInput = "D:\\hadoop_study_data\\topN\\input";
        String countOutput = "D:\\hadoop_study_data\\topN\\output3";
        String topNInput = "D:\\hadoop_study_data\\topN\\output3\\part-r-00000";
        String topNOutput = "D:\\hadoop_study_data\\topN\\output4";

        Configuration configuration = new Configuration();
        Job job_count = Job.getInstance(configuration, "count");
        job_count.setJarByClass(CombineDriver.class);
        job_count.setMapperClass(CountMapper.class);
        job_count.setReducerClass(CountReducer.class);

        job_count.setMapOutputKeyClass(Text.class);
        job_count.setMapOutputValueClass(LongWritable.class);

        job_count.setOutputKeyClass(Text.class);
        job_count.setOutputValueClass(LongWritable.class);

        FileInputFormat.setInputPaths(job_count, new Path(countInput));
        FileOutputFormat.setOutputPath(job_count, new Path(countOutput));

        Job job_topN = Job.getInstance(configuration, "topN");
        job_topN.setJarByClass(CombineDriver.class);
        job_topN.setMapperClass(TopNMapper.class);
        job_topN.setReducerClass(TopNReducer.class);

        job_topN.setMapOutputKeyClass(ViewCountBean.class);
        job_topN.setMapOutputValueClass(Text.class);

        job_topN.setOutputFormatClass(ViewCountOutputFormat.class);
        job_topN.setOutputKeyClass(Text.class);
        job_topN.setOutputValueClass(ViewCountBean.class);

        FileInputFormat.setInputPaths(job_topN, new Path(topNInput));
        FileOutputFormat.setOutputPath(job_topN, new Path(topNOutput));

        boolean result = job_count.waitForCompletion(true);
        if(result){
            boolean topN_result = job_topN.waitForCompletion(true);
            System.exit(topN_result ? 0 : 1);
        }

    }
}
